import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.SimpleDateFormat;
import java.util.*;

public class stuInfo {
    private static final char[] ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ".toCharArray();
    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd");
    private static Set<String> usedNames = new HashSet<>(); // 记录已使用的名字

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.235.128:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        try {
            while (true) {
                try {
                    for (int i = 0; i < 5; i++) {
                        Random random = new Random();

                        // 生成班级号（这里简单示例为1到10的随机数）
                        int classNumber = random.nextInt(10) + 1;
                        String className = "Class" + classNumber;

                        // 生成不重复的随机名字（3-6位，首字母大写）
                        String name = generateUniqueName(random);

                        // 生成性别 1 为男 0 为女
                        int gender = random.nextInt(2);

                        // 生成出生年月日（这里简单示例为固定格式的随机日期）
                        int year = random.nextInt(2023 - 2018 + 1) + 2018;
                        int month = random.nextInt(12) + 1;
                        int day = random.nextInt(28) + 1;
                        String birthDate = year + "-" + month + "-" + day;

                        // 生成学号（这里简单示例为10000到99999的随机数）
                        int studentId = random.nextInt(90000) + 10000;

                        // 生成学期 1 ~ 8
                        int semester = random.nextInt(4) + 1;

                        // 生成绩点(1 - 10)，随机生成两个绩点并用逗号连接
                        int gpa1 = random.nextInt(10) + 1;
                        int gpa2 = random.nextInt(10) + 1;
                        String gpa = gpa1 + "," + gpa2;

                        // 生成在校状态:是否在校 L是毕业，A 在校
                        String status = random.nextInt(2) == 0 ? "L" : "A";

                        // 按照要求拼接数据，用制表符\t作为分隔符
                        String mockData = className + "\t" + name + "\t" + gender + "\t" + birthDate + "\t" + studentId + "\t" + semester + "\t" + gpa + "\t" + status;
                        System.out.println("" + mockData);
                        kafkaProducer.send(new ProducerRecord<>("stuInfo", mockData));
                    }

                    Thread.sleep(5000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } finally {
            kafkaProducer.close();
        }
    }

    // 生成唯一的随机名字
    private static String generateUniqueName(Random random) {
        while (true) {
            // 随机生成3-6位长度
            int length = random.nextInt(4) + 3;
            StringBuilder sb = new StringBuilder();

            // 首字母大写
            sb.append(ALPHABET[random.nextInt(26)]);

            // 后续字母小写
            for (int i = 1; i < length; i++) {
                sb.append((char) (random.nextInt(26) + 'a'));
            }

            String name = sb.toString();

            // 检查是否重复
            if (!usedNames.contains(name)) {
                usedNames.add(name);
                // 当已使用名字超过1000个时，清空集合避免内存溢出
                if (usedNames.size() > 1000) {
                    usedNames.clear();
                }
                return name;
            }
        }
    }
}